package edu.nf.redis_retry.service;

import com.rabbitmq.client.Channel;
import edu.nf.redis_retry.config.RabbitConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

import java.io.IOException;

@Service
@Slf4j
@RequiredArgsConstructor
public class ConsumerService {

    private final RedisTemplate<String, String> redisTemplate;

    /** 最大重试次数 */
    private static final int MAX_RETRY = 3;
    /** Redis key 前缀 */
    private static final String RETRY_KEY_PREFIX = "retry:count:";

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = RabbitConfig.QUEUE_NAME, durable = "false",
                    arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = RabbitConfig.DEAD_EXCHANGE_NAME),
                            @Argument(name = "x-dead-letter-routing-key", value = RabbitConfig.DEAD_ROUTING_KEY)
                    }),
            exchange = @Exchange(name = RabbitConfig.EXCHANGE_NAME, type = ExchangeTypes.DIRECT),
            key = RabbitConfig.ROUTING_KEY
    ), ackMode = "MANUAL")

    // 显式声明手动签收
    public void receiveMessage(String message,
                               Channel channel,
                               @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                               @Header(name = "messageId", required = false) String messageId) throws IOException {

        // 1. 生成唯一标识（若发送端未设 messageId，可用消息内容 MD5 代替）
        String uniqueKey = messageId == null ? "msg:" + message.hashCode() : messageId;
        String redisKey = RETRY_KEY_PREFIX + uniqueKey;

        try {
            log.info("接收到消息: {}", message);

            // 业务逻辑
            if ("自定义异常".equals(message)) {
                throw new RuntimeException("自定义异常");
            }
            //业务结束

            // 2. 成功 -> 签收并清理 Redis 计数
            redisTemplate.delete(redisKey);
            channel.basicAck(deliveryTag, false);
            log.info("消息消费成功，已 ACK");

        } catch (Exception e) {
            log.error("消息消费失败: {}", e.getMessage());

            // 失败 -> Redis 计数
            Long retryCount = redisTemplate.opsForValue().increment(redisKey);
            redisTemplate.expire(redisKey, java.time.Duration.ofMinutes(10)); // 防脏数据

            if (retryCount != null && retryCount < MAX_RETRY) {
                // 未达到上限，重新入队
                log.warn("第 {} 次重试", retryCount);
                channel.basicNack(deliveryTag, false, true);
            } else {
                // 达到上限，进入死信
                log.warn("已重试 {} 次，进入死信队列", MAX_RETRY);
                channel.basicNack(deliveryTag, false, false);
            }
        }
    }
}